package com.qth.valet.common.queue;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.Redisson;
import org.redisson.RedissonShutdownException;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;

/**
 * 初始化队列监听
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class RedisDelayedQueueInit implements ApplicationContextAware {

    private final RedissonClient redissonClient;

    /**
     * 获取应用上下文并获取相应的接口实现类
     */
    @Override
    @SuppressWarnings("rawtypes")
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        Map<String, RedisDelayedQueueListener> beansMap = applicationContext.getBeansOfType(RedisDelayedQueueListener.class);
        for (Map.Entry<String, RedisDelayedQueueListener> taskEventListenerEntry : beansMap.entrySet()) {
            // 以类名作为队列名称
            String listenerName = taskEventListenerEntry.getValue().getClass().getName();
            // 开启一个线程监听
            startThread(listenerName, taskEventListenerEntry.getValue());
        }
    }

    /**
     * 启动线程获取队列
     */
    @SuppressWarnings({"rawtypes", "unchecked"})
    private <T> void startThread(String queueName, RedisDelayedQueueListener redisDelayedQueueListener) {
        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(queueName);
        //服务重启后，无offer，take不到信息。
        redissonClient.getDelayedQueue(blockingFairQueue);
        //由于此线程需要常驻，可以新建线程，不用交给线程池管理
        Thread thread = new Thread(() -> {
            log.info("启动监听队列线程: {}", queueName);
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    T t = blockingFairQueue.take();
                    log.info("监听队列线程，监听名称：{}, 内容: {}", queueName, t);
                    redisDelayedQueueListener.invoke(t);
                } catch (RedissonShutdownException | InterruptedException e) {
                    log.info("监听队列线程错误，监听线程中断，将无法再监听关单。",  e);
                    Thread.currentThread().interrupt();
                }
            }
        });
        thread.setName(queueName);
        thread.start();
    }

}
